package streaming

import java.sql.DriverManager

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 从网络读取数据进行统计，存储到MySQL
  * 在控制台输入
  * nc -lk 9999
  * 创建表：
  * create table tb_word_count(
  * word varchar(50) default null,
  * word_count int(10) default null
  * );
  */
object ForeachRDDApp {

  def main(args: Array[String]): Unit = {
    val sc = new SparkConf().setMaster("local[*]").setAppName("ForeachRDDApp")

    val ssc = new StreamingContext(sc, Seconds(5))

    // 如果使用stateful的算子，必须要设置checkpoint
    // 生产环境中，checkpoint设置在HDFS某个文件夹中
    //ssc.checkpoint(".")

    val lines = ssc.socketTextStream("localhost", 9999)

    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)

    //结果写入MySQL
    // 方案一：connection在driver端创建，在worker端执行
    //    result.foreachRDD { rdd =>
    //      val connection = createConnection()  // executed at the driver
    //      rdd.foreach { record =>
    //        connection.send(record) // executed at the worker
    //      }
    //    }

    // 方案二：存在反复创建connection的问题
    //    result.foreachRDD { rdd =>
    //      rdd.foreach { record =>
    //        val connection = createConnection()
    //        connection.send(record)
    //        connection.close()
    //      }
    //    }

    // 方案三：在Partition中创建连接
    result.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
        // 在partition中创建connection
        val connection = createConnection();
        partitionOfRecords.foreach(record => {
          val sql = "insert into tb_word_count(word,word_count) values('" + record._1 + "'," + record._2 + ")"
          println(sql)
          // 存储数据
          connection.createStatement().execute(sql)
        })
        connection.close()
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }

  /**
    * 获取MySQL连接
    * 可以改为连接池
    *
    * @return
    */
  def createConnection() = {
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
  }
}
